package geektime.spring.data.reactive.simpler2dbcdemo;

import geektime.spring.data.reactive.simpler2dbcdemo.converter.MoneyReadCoverter;
import geektime.spring.data.reactive.simpler2dbcdemo.converter.MoneyWriteConverter;
import geektime.spring.data.reactive.simpler2dbcdemo.model.Coffee;
import io.r2dbc.h2.H2ConnectionConfiguration;
import io.r2dbc.h2.H2ConnectionFactory;
import io.r2dbc.spi.ConnectionFactory;

import java.util.Arrays;
import java.util.concurrent.CountDownLatch;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.data.convert.CustomConversions;
import org.springframework.data.convert.CustomConversions.StoreConversions;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration;
import org.springframework.data.r2dbc.dialect.Dialect;
import org.springframework.data.r2dbc.function.DatabaseClient;
import org.springframework.data.r2dbc.function.convert.R2dbcCustomConversions;

@Slf4j
@SpringBootApplication
public class SimpleR2dbcDemoApplication
        extends AbstractR2dbcConfiguration
        implements ApplicationRunner {

    @Autowired
    private DatabaseClient databaseClient;

    public static void main(String[] args) {
        SpringApplication.run(SimpleR2dbcDemoApplication.class, args);
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        CountDownLatch cdl = new CountDownLatch(2);

        databaseClient.execute().sql("select * from t_coffee").as(Coffee.class).fetch().first().doFinally(s -> cdl.countDown())
                //        .subscribeOn(Schedulers.elastic())
                .subscribe(c -> log.info("Fetch execute(){}", c));

        databaseClient.select().from("t_coffee").orderBy(Sort.by(Direction.DESC, "id")).page(PageRequest.of(0, 3)).as(Coffee.class).fetch().all().doFinally(s -> cdl.countDown())
                //        .subscribeOn(Schedulers.elastic())
                .subscribe(c -> log.info("Fetch select (){}", c));

        log.info("After starting.");
        cdl.countDown();
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        return new H2ConnectionFactory(H2ConnectionConfiguration.builder().inMemory("testdb").username("sa").build());
    }

    @Bean
    public R2dbcCustomConversions r2dbcCustomConversions() {
        Dialect dialect = getDialect(connectionFactory());
        CustomConversions.StoreConversions storeConversions = StoreConversions.of(dialect.getSimpleTypeHolder());
        return new R2dbcCustomConversions(storeConversions, Arrays.asList(new MoneyReadCoverter(), new MoneyWriteConverter()));
    }

}













